pub mod ldata;
pub mod ltopic;

use chrono::{DateTime, Utc};
use lazy_static::lazy_static;
use rumqttc::v5::mqttbytes::v5::{Packet, Publish};
use rumqttc::v5::{
    AsyncClient, Client, ConnectionError, Event, MqttOptions, StateError, mqttbytes::QoS,
};
use std::collections::HashMap;
use std::io::ErrorKind;
use std::sync::atomic::{AtomicBool, AtomicU16, Ordering};
use std::sync::{Arc, LazyLock};
use std::time::{Duration, SystemTime};
use tokio::sync::{Mutex, OnceCell, mpsc, oneshot, watch};
// 需要在 Cargo.toml 中添加 chrono 依赖
// $SYS/broker/clients/connected：当前连接数
// $SYS/broker/clients/total：总客户端数（包括离线）
// $SYS/broker/clients/active：活跃客户端数
// 客户端断开：
// $SYS/broker/clients/disconnected：断开连接数
// $SYS/broker/clients/expired：会话过期数
// 具体客户端状态：
// $SYS/broker/connection/<client-id>/state：特定客户端连接状态
// $SYS/broker/connection/<client-id>/uptime：客户端在线时间

// 定义

// 命令枚举
#[derive(Debug)]
pub enum MqttCommand {
    Subscribe {
        topic: String,
        qos: QoS,
        response: oneshot::Sender<Result<(), String>>,
    },
    Publish {
        topic: String,
        mess_id: String,
        payload: Vec<u8>,
        qos: QoS,
        response: oneshot::Sender<Result<(), String>>,
    },
    Disconnect,
    Connect,
}

#[derive(Debug, Default, Clone)]
pub struct MqttMsg {
    topic: String,
    msg: String,
}

const MQTT_HOST: &str = "localhost";
const MQTT_PORT: u16 = 1883;

// mqtt 的 sender
static SENDER_MQTT_CLIENT: LazyLock<OnceCell<mpsc::Sender<MqttCommand>>> =
    LazyLock::new(|| OnceCell::new());
// mqtt 的 watch
pub static WATCH_MQTT_MSG: LazyLock<OnceCell<watch::Receiver<MqttMsg>>> =
    LazyLock::new(|| OnceCell::new());

// 全局状态标记：false表示未执行，true表示已执行
static HAS_RUN: AtomicBool = AtomicBool::new(false);
static POS_PACKT: AtomicU16 = AtomicU16::new(0);

pub async fn create_mqtt(
    id: String,
    username: String,
    password: String,
) -> Result<(), anyhow::Error> {
    let has_run = HAS_RUN
        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
        .is_ok();

    if !has_run {
        // 不是第一次执行
        return Ok(());
    }
    let hash_msg_packet: Arc<Mutex<HashMap<u16, String>>> = Arc::new(Mutex::new(HashMap::new()));
    let hash_msg_packet_clone = hash_msg_packet.clone();
    let client_id_clone = id.clone();
    let mut mqttoptions = MqttOptions::new(id, MQTT_HOST, MQTT_PORT);

    // 配置选项
    mqttoptions
        .set_keep_alive(Duration::from_secs(60)) // 保活间隔
        .set_clean_start(false)
        .set_session_expiry_interval(Some(u32::MAX))
        .set_credentials(username, password) // 认证信息
        .set_max_packet_size(Some(1024 * 1024)) // 最大包大小(1MB)
        .set_outgoing_inflight_upper_limit(1000);

    let (client, mut event_loop) = AsyncClient::new(mqttoptions, 10); // 10是通知通道容量

    // client.subscribe("#", QoS::AtMostOnce).await?;
    // client.subscribe("$SYS/broker/clients/connected", QoS::AtMostOnce).await?;
    // match client.subscribe("$SYS/#", QoS::AtMostOnce).await {
    //     Ok(_) => {},
    //     Err(err) => {
    //         println!("err: {:?}", err);
    //     }
    // }
    // client.subscribe("#", QoS::AtMostOnce).await?;
    client
        .subscribe(
            format!("{}/{}", ltopic::MQTT_TOPIC_HEAD_MSG, &client_id_clone),
            QoS::AtMostOnce,
        )
        .await?;

    println!("已订阅系统主题，等待消息...");
    let (command_sender, mut command_receiver) = mpsc::channel(100);
    // sender
    SENDER_MQTT_CLIENT
        .get_or_init(|| async { command_sender })
        .await;

    // 启动事件处理任务
    tokio::spawn(async move {
        while let Some(command) = command_receiver.recv().await {
            let dt: DateTime<Utc> = SystemTime::now().into();
            println!("{:?}, 接受的reveiver： {:?}", dt, command);
            match command {
                MqttCommand::Subscribe {
                    topic,
                    qos,
                    response,
                } => {
                    let result = client.subscribe(&topic, qos).await;
                    let _ = response.send(result.map_err(|e| {
                        println!("发送消息错误：{:?}", e);
                        e.to_string()
                    }));
                }
                MqttCommand::Publish {
                    topic,
                    mess_id,
                    payload,
                    qos,
                    response,
                } => {
                    let packet_id = get_packet_id();
                    hash_msg_packet_clone
                        .lock()
                        .await
                        .insert(packet_id, mess_id);
                    println!("topic_id: {}", packet_id);
                    let result = client.handle_publish_l(&topic,packet_id, qos, false, payload, None).await;
                    // let result = client.publish(&topic, qos, false, payload).await;
                    let _ = response.send(result.map_err(|e| e.to_string()));
                }
                MqttCommand::Disconnect => {
                    let r = client.disconnect().await;
                }
                MqttCommand::Connect => {}
            }
        }
    });

    let (rx, tx) = watch::channel(MqttMsg::default());

    WATCH_MQTT_MSG.get_or_init(|| async { tx }).await;

    loop {
        let pool_rs = event_loop.poll().await;
        let dt: DateTime<Utc> = SystemTime::now().into();
        println!("{:?}, --------pool_rs: {:?}", dt, pool_rs);
        match pool_rs {
            Ok(Event::Incoming(pack)) => match pack {
                Packet::Publish(publish) => {
                    handle_system_message(&publish, &rx).await;
                }
                Packet::PubAck(p) => {
                    let mess_id = hash_msg_packet.lock().await.get(&p.pkid);
                    let reson = p.reason;
                }
                Packet::Connect(mconn, mwill, mlogin) => {}
                Packet::Disconnect(dis) => {}
                _ => {}
            },
            Ok(Event::Outgoing(event)) => {}
            Err(err) => match err {
                ConnectionError::Io(err) => {
                    println!("mqtt IO错误，err: {:?}", err);
                }
                ConnectionError::ConnectionRefused(code) => {}
                ConnectionError::MqttState(state) => match state {
                    _ => {}
                },
                _ => {
                    println!("mqtt错误，err: {:?}", err);
                }
            },
        }
    }
    Ok(())
}

async fn handle_system_message(publish: &Publish, rx: &watch::Sender<MqttMsg>) {
    let payload_str = String::from_utf8_lossy(&publish.payload).into_owned();
    let topic_str = String::from_utf8_lossy(&publish.topic).into_owned();

    println!("topic_str:{}, payload_str: {}, ", &topic_str, &payload_str);

    rx.send(MqttMsg {
        topic: topic_str,
        msg: payload_str,
    })
    .expect("");
}

pub async fn disconnect() {
    if let Some(sender) = SENDER_MQTT_CLIENT.get() {
        sender.send(MqttCommand::Disconnect).await.unwrap();
    }
}

pub async fn connect() {
    if let Some(sender) = SENDER_MQTT_CLIENT.get() {
        sender.send(MqttCommand::Connect).await.unwrap();
    }
}

// 订阅方法
pub async fn subscribe(topic: &str, qos: u8) -> Result<(), String> {
    let qos_ = match qos {
        0 => QoS::AtMostOnce,
        1 => QoS::AtLeastOnce,
        2 => QoS::ExactlyOnce,
        _ => QoS::AtMostOnce,
    };
    let (response_sender, response_receiver) = oneshot::channel();
    if let Some(sender) = SENDER_MQTT_CLIENT.get() {
        sender
            .send(MqttCommand::Subscribe {
                topic: topic.to_string(),
                qos: qos_,
                response: response_sender,
            })
            .await
            .map_err(|e| e.to_string())?;
    }
    response_receiver.await.expect("接收响应失败")
}

// 发布方法
pub async fn publish(
    topic: String,
    mess_id: String,
    payload: Vec<u8>,
    qos: u8,
) -> Result<(), String> {
    let qos_ = match qos {
        0 => QoS::AtMostOnce,
        1 => QoS::AtLeastOnce,
        2 => QoS::ExactlyOnce,
        _ => QoS::AtMostOnce,
    };
    let (response_sender, response_receiver) = oneshot::channel();
    if let Some(sender) = SENDER_MQTT_CLIENT.get() {
        println!("发送消息了");
        sender
            .send(MqttCommand::Publish {
                topic,
                mess_id,
                payload,
                qos: qos_,
                response: response_sender,
            })
            .await
            .map_err(|e| e.to_string())?;
    }
    response_receiver.await.expect("接收响应失败")
}

pub fn get_packet_id() -> u16 {
    let old_next = POS_PACKT.fetch_add(1, Ordering::Relaxed);
    if old_next == 0 {
        POS_PACKT.fetch_add(1, Ordering::Relaxed);
        return 1;
    }
    old_next + 1
}


